package jsen.tool.ddd.adapter.event.common;

import jsen.tool.ddd.adapter.event.AsyncCallback;
import jsen.tool.ddd.adapter.event.Consumer;
import jsen.tool.ddd.adapter.event.DomainEvent;
import jsen.tool.ddd.adapter.event.IEventBus;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @author jsen
 * @date 2021/10/6
 */
public class EventBus implements IEventBus, Dispatcher {

    private final List<Consumer<? extends DomainEvent>> consumerList;
    private final ExecutorService executor;

    public EventBus(ExecutorService executor) {
        this.executor = executor;
        consumerList = new ArrayList<>(16);
    }

    @Override
    public <E extends DomainEvent> void fireEventAsync(E event, AsyncCallback<E> callback) {
        executor.execute(() -> {
            fireEvent(event);
            callback.succeed(event);
        });
    }

    @Override
    public <E extends DomainEvent> void fireEventParallel(E event, AsyncCallback<E> callback) {
        dispatchParallel(event);
        callback.succeed(event);
    }

    @Override
    public <E extends DomainEvent> void fireEvent(E event) {
        dispatch(event);
    }

    @Override
    public <E extends DomainEvent> void register(Consumer<E> consumer) {
        consumerList.add(consumer);
    }


    @Override
    @SuppressWarnings("unchecked")
    public <X extends DomainEvent, E extends DomainEvent> void dispatch(E event) {
        for (Consumer<? extends DomainEvent> consumer : consumerList) {
            Consumer<X> xConsumer = (Consumer<X>) consumer;
            if (consumer.match(event)) {
                xConsumer.accept((X) event);
            }
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public <X extends DomainEvent, E extends DomainEvent> void dispatchParallel(E event) {
        List<Consumer<? extends DomainEvent>> matchList = consumerList.stream().filter(i -> i.match(event))
                .collect(Collectors.toList());
        SyncTask<Consumer<? extends DomainEvent>> task = new SyncTask<Consumer<? extends DomainEvent>>(executor);
        task.addTask(matchList, c -> {
            Consumer<X> xConsumer = (Consumer<X>) c;
            xConsumer.accept((X) event);
        });

        task.sync();
    }
}
